package io.kiki.sba.registry.server.clientInterface.client.manager;

import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.api.exchange.message.SimpleRequest;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.sessionserver.CheckClientManagerRequest;
import io.kiki.sba.registry.common.model.sessionserver.CheckClientManagerResponse;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.providedata.FetchClientOffAddressService;
import io.kiki.sba.registry.server.clientInterface.resource.Sdks;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.store.api.meta.ClientManagerAddressRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


public class CheckClientManagerService {
    private static final Logger logger = LoggerFactory.getLogger(CheckClientManagerService.class);

    private static final long NO_NEED_CHECK = 0L;
    private final Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder().retryIfException().retryIfResult(input -> !input).withWaitStrategy(WaitStrategies.fixedWait(200, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(15)).build();
    @Autowired
    private ExecutorManager executorManager;
    @Resource
    private FetchClientOffAddressService fetchClientOffAddressService;
    @Autowired
    private ClientManagerAddressRepository clientManagerAddressRepository;
    @Autowired
    private Config config;
    @Autowired
    private MetaServerService metaServerService;
    @Autowired
    private NodeExchanger sessionConsoleExchanger;

    /**
     * check local version and other session version
     */
    public boolean check(long expectedVersion) {

        if (expectedVersion == NO_NEED_CHECK) {
            return true;
        }
        Future<Tuple<Boolean, Long>> future = executorManager.getClientManagerCheckExecutor().submit(() -> checkLocalCache(expectedVersion));

        List<Url> servers = getOtherConsoleServersCurrentZone();
        logger.info("checkInZone expectedVersion: {}, others={}", expectedVersion, servers);
        if (servers.size() == 0) {
            return checkLocalResp(expectedVersion, future);
        }
        Map<Url, Response> map = Sdks.concurrentSdkSend(executorManager.getClientManagerCheckExecutor(), servers, (Url url) -> {
            final CheckClientManagerRequest request = new CheckClientManagerRequest(expectedVersion);
            return (Response) sessionConsoleExchanger.request(new SimpleRequest(request, url)).getResult();
        }, 3000);

        return checkLocalResp(expectedVersion, future) && checkOtherServersResp(map, expectedVersion);
    }


    private boolean checkOtherServersResp(Map<Url, Response> map, long expectedVersion) {
        for (Entry<Url, Response> entry : map.entrySet()) {
            if (!(entry.getValue() instanceof GenericResponse)) {
                logger.error("url={} client manager check fail, expectedVersion: {}, resp: {}", entry.getKey().getIp(), expectedVersion, entry.getValue());
                return false;
            }
            GenericResponse<CheckClientManagerResponse> response = (GenericResponse) entry.getValue();
            CheckClientManagerResponse data = response.getData();
            if (!response.isSuccess() || !data.isPaasCheck()) {
                logger.error("url={} client manager check fail, expectedVersion: {}, actual: {}", entry.getKey().getIp(), expectedVersion, data.getActualVersion());
                return false;
            }
        }
        return true;
    }


    public Tuple<Boolean, Long> checkLocalCache(long expectedVersion) {

        final AtomicLong version = new AtomicLong(0);
        try {
            retryer.call(() -> {
                clientManagerAddressRepository.wakeup();
                fetchClientOffAddressService.wakeup();
                version.set(fetchClientOffAddressService.lastLoadVersion());
                return version.get() >= expectedVersion;
            });
            return new Tuple(true, version.get());
        } catch (Throwable t) {
            logger.error("[checkLocalCache]client manager check version, excepted: {}, actual: {} error.", expectedVersion, version.get(), t);
            return new Tuple(false, version.get());
        }
    }


    private boolean checkLocalResp(long expectedVersion, Future<Tuple<Boolean, Long>> submit) {
        try {
            Tuple<Boolean, Long> tuple = submit.get(3100, TimeUnit.MILLISECONDS);
            if (!paasCheck(tuple.o1, false)) {
                logger.error("client manager local check fail, expectedVersion: {}, actual: {}", expectedVersion, tuple.o2);
                return false;
            }
            return true;
        } catch (Throwable t) {
            logger.error("client manager local check error. expectedVersion: {}", expectedVersion, t);
            return false;
        }
    }

    private List<Url> getOtherConsoleServersCurrentZone() {
        return Sdks.getOtherConsoleServers(null, config, metaServerService);
    }

    private boolean paasCheck(Boolean bool, boolean defaultValue) {
        return bool == null ? defaultValue : bool.booleanValue();
    }
}
